package com.atguigu.ting.realtime.app.dwd.db;

import com.atguigu.ting.realtime.app.BaseSqlApp;
import com.atguigu.ting.realtime.common.Constant;
import com.atguigu.ting.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/4/12 09:23
 */
public class DwdInteractionTrackComment extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdInteractionTrackComment().init(
            3008,
            2,
            "DwdInteractionTrackComment"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db数据
        readOdsDb(tEnv, "DwdInteractionTrackComment");
        // 2. 过滤出专辑评论数据
        Table albumComment = tEnv.sqlQuery(
            "select " +
                "`data`['id'] id, " +
                "`data`['album_id'] album_id, " +
                "`data`['track_id'] track_id, " +
                "`data`['user_id'] user_id, " +
                "`data`['content'] content, " +
                "`data`['create_time'] comment_time," +
                "ts " +
                "from ods_db " +
                "where `database`='ting' " +
                "and `table`='track_comment' " +
                "and (`type`='insert' or `type`='update')  ");
        // 3. 写出到 kafka
        tEnv.executeSql(
            "create table dwd_interaction_tack_comment(" +
                "id string, " +
                "album_id string, " +
                "track_id string, " +
                "user_id string, " +
                "content string, " +
                "comment_time string," +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_INTERACTION_TRACK_COMMENT));
    
        albumComment.executeInsert("dwd_interaction_tack_comment");
    }
}
